%%% -------------------------------------------------------------------
%%% Author  : mjollnir.ray
%%% Description :
%%%
%%% Created : 2009-3-18
%%% -------------------------------------------------------------------
-module(node_store).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include("ring.hrl").
-include("map.hrl").

%% --------------------------------------------------------------------
%% External exports
-export([start_link/1]).
-export([send/2, get/1, put/2]).
-export([acceptor_loop/1]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

-define(STORE_SERVER, ?MODULE).
-define(WAIT_TABLE_TIMEOUT, 30000).

-record(node_store, {
        ring_node :: ring_node(),
%        port = 8899,
        max = 2048,
%        ip = any,
        listen = null,
        acceptor = null                
    }).
-record(gis_data, {
        key = 0 :: non_neg_integer(),
        value :: cell()
    }).

-define(IDLE_TIMEOUT, 30000).

%% ====================================================================
%% External functions
%% ====================================================================
start_link(RingNode) ->
    log4erl:debug("node_store: ~p", [RingNode]),
    State = #node_store{ring_node = RingNode},
    gen_server:start_link({local, ?STORE_SERVER}, ?STORE_SERVER, State, []).

%% ====================================================================
%% Server functions
%% ====================================================================
send(From, Key) ->
    log4erl:debug("Key: ~p", [Key]),
%    Value = mnesia:dirty_read(gis_data, Key),
    From ! "Value",
    timer:sleep(10*1000),
    send_ok.

-spec get(Key :: non_neg_integer()) -> cell().
get(Key) ->
    case mnesia:dirty_read(gis_data, Key) of
        [#gis_data{key=Key, value=Value}] ->
            Value;
        [] ->
            nil
    end.

-spec put(Key :: non_neg_integer(), Value::cell()) ->
    {'ok', 'success'} | {'error', any()}.
put(Key, Value) ->
    F = fun() ->
        mnesia:write(#gis_data{key = Key, value = Value})    
    end,
    case mnesia:transaction(F) of
    {atomic, _} ->
        log4erl:debug("store value for: ~p", [Key]),
        ok;
    {aborted, R} ->
        log4erl:fail("aborted reason:~p~n", [R]),
        {error, aborted}
    end.

%% --------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          ignore               |
%%          {stop, Reason}
%% --------------------------------------------------------------------
init(State=#node_store{ring_node=#ring_node{access_port=Port}}) ->
    log4erl:debug("DB Start: ~p", [node()]),
    process_flag(trap_exit, true),

    case mnesia:create_schema([node()]) of
        {error,{_,{already_exists,_}}} ->
            log4erl:debug("has already exist db~n");
        ok ->
            ok
    end,
    mnesia:start(),
    init_db(),
 
    Res = gen_tcp_listen(Port, State),
    ring_lord:add_ringnode(State#node_store.ring_node),
    Res.
%    {ok, #node_store{}}.

gen_tcp_listen(Port, State) ->
    log4erl:debug("gen tcp listen on port: ~p", [Port]),
    BaseOpts = [binary, 
                {reuseaddr, true},
                {packet, 2},
%                {backlog, Backlog},
%                {recbuf, 8192},
                {active, false},
                {nodelay, true}],
    case gen_tcp:listen(Port, BaseOpts) of
        {ok, Listen} ->
            {ok, ListenPort} = inet:port(Listen),
            log4erl:debug("listen Port: ~p", [ListenPort]),
            RingNode = State#node_store.ring_node,
            RingNode1 = RingNode#ring_node{access_port=ListenPort},
            {ok, new_acceptor(State#node_store{listen=Listen, ring_node=RingNode1})};
        {error, eaddrinuse} ->
            log4erl:fatal("TCP port(~p) is in used by other app", [Port]),
            {stop, eaddrinuse};
        {error, Reason} ->
            {stop, Reason}
    end.

new_acceptor(State=#node_store{max=0}) ->
    log4erl:warn("Not accepting new connections~n"),
    State#node_store{acceptor=null};
new_acceptor(State=#node_store{listen=Listen}) ->
    log4erl:debug("start new acceptor"),
    Pid = proc_lib:spawn_link(?MODULE, acceptor_loop, [{self(), Listen}]),
    State#node_store{acceptor=Pid}.

acceptor_loop({Server, Listen}) ->
    log4erl:debug("acceptor_loop"),
    case catch gen_tcp:accept(Listen) of
        {ok, Socket} ->
            gen_server:cast(Server, {accepted, self()}),
            recv_request(Socket);
        {error, closed} ->
            exit({error, closed});
        Other ->
            log4erl:fatal("Accept failed error: ~p", [Other]),
            exit({error, accept_failed})
    end.
    
%% --------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_call(Request, _From, State) ->
    log4erl:debug("Req: ~p", [Request]),
    Reply = (lists:reverse(Request)),
    {reply, Reply, State}.

%% --------------------------------------------------------------------
%% Function: handle_cast/2
%% Description: Handling cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_cast({accepted, Pid}, State=#node_store{acceptor=Pid, max=Max}) ->
    State1 = State#node_store{max = Max - 1},
    {noreply, new_acceptor(State1)};
handle_cast(_Msg, State) ->
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: handle_info/2
%% Description: Handling all non call/cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_info({'EXIT', Pid, normal},
            State=#node_store{acceptor=Pid}) ->
    log4erl:debug("normal acceptor down~n"),
    {noreply, new_acceptor(State)};
handle_info({'EXIT', Pid, Reason},
            State=#node_store{acceptor=Pid}) ->
    log4erl:error("acceptor exit abnormal: ~p", [Reason]),
    timer:sleep(100),
    {noreply, new_acceptor(State)};
handle_info({'EXIT', _ExitPid, Reason},
            State=#node_store{acceptor=Pid, max=Max}) ->
    case Reason of 
        normal ->
            log4erl:debug("ok: acceptor pid normal exit."),
            ok;
        _ ->
            log4erl:error("child error: ~p", [Reason])
    end,
    State1 = State#node_store{max=Max + 1},
    State2 = case Pid of
                 null ->
                     new_acceptor(State1);
                 _ ->
                     State1
             end,
    {noreply, State2};
handle_info(_Info, State) ->
    log4erl:info("handle_info: ~p", [_Info]),
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: terminate/2
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%% --------------------------------------------------------------------
terminate(_Reason, _State) ->
    ok.

%% --------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
%% Returns: {ok, NewState}
%% --------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% --------------------------------------------------------------------
%%% Internal functions
%% --------------------------------------------------------------------
init_db() ->
    Tables = mnesia:system_info(tables),
    case lists:member(gis_data, Tables) of
        true ->
            ok;
    false ->
        case catch create_tables() of
            ok ->
                log4erl:debug("created table ok~n", []);
            Error ->
                log4erl:debug("create table error ~p~n", [Error])
        end
    end,

    ok = wait_for_tables().
    
create_tables() ->
    lists:foreach(fun({Tab, TabArrt}) ->
                    case mnesia:create_table(Tab, TabArrt) of
                        {atomic, ok} -> ok;
                        {aborted, Reason} ->
                            throw({error, Reason})
                    end
                end,
                table_definitions()),
    ok.

wait_for_tables() ->
    case mnesia:wait_for_tables(table_names(), ?WAIT_TABLE_TIMEOUT) of
        ok -> ok;
        {timeout, BadTabs} ->
            throw({error, {timeout_waiting_for_tables, BadTabs}});
        {error, Reason} ->
            throw({error, {failed_waiting_for_tables, Reason}})
    end.
    
table_definitions() ->
    [{gis_data, [{type, ordered_set}, {disc_copies, [node()]}, {attributes, record_info(fields, gis_data)}]}
    ].

table_names() ->
    [Tab || {Tab, _} <- table_definitions()].

recv_request(Socket) ->
    log4erl:debug("receive request"),
    case gen_tcp:recv(Socket, 0, ?IDLE_TIMEOUT) of
        {ok, Bin} ->
            handle_request(Socket, Bin);
        {error, closed} ->
            gen_tcp:close(Socket),
            exit(normal)
    end.

handle_request(Socket, Bin) ->
    R = binary_to_term(Bin),
    do_store(R, Socket).

do_store({put, Key, #cell{} = Value}, Socket) ->
    ?MODULE:put(Key, Value),
    gen_tcp:send(Socket, term_to_binary(Key));
do_store({get, Key}, Socket) ->
    Value = ?MODULE:get(Key),
    ReplyBin = term_to_binary(Value),
    gen_tcp:send(Socket, ReplyBin).